package com.lq.rocketmq.order;

import com.lq.rocketmq.config.MyConstant;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName OrderProducer
 * @Description 顺序消费生产者
 * @Author LQ
 * @Date 2021/5/15 23:17
 * @Version 1.0
 */
public class OrderProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer,并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer(MyConstant.MSG_GROUP);
        //2.指定Nameserver地址
        producer.setNamesrvAddr(MyConstant.NAMESRV_ADDR);
        //3.启动producer
        producer.start();
        // 构建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();
        // 发送消息
        for (int i = 0; i < orderSteps.size(); i++) {
            OrderStep orderStep = orderSteps.get(i);
            Message msg = new Message(MyConstant.TOPIC, MyConstant.TAG1, "" + i, orderStep.toString().getBytes());
            /**
             * 发送消息
             * 这里设置三个参数
             * 1:消息对象
             * 2.消息队列选择
             * 3.选择队列的业务标识(订单id)
             */
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                /**
                 *
                 * @param messageQueues 队列集合
                 * @param message 消息对象
                 * @param arg 业务标识参数(订单id)
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> messageQueues, Message message, Object arg) {
                    // arg为发送的时候传递的参数标识,这里为订单id
                    Long orderId = (Long) arg;
                    // 取模,保证是同一个消息队列(list为队列集合,broker中的队列集合,)保证同一个订单在同一个队列中
                    long index = orderId % messageQueues.size();
                    return messageQueues.get((int) index);
                }
            }, orderStep.getOrderId());
            System.out.println("发送结果:" + sendResult);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}
